DequeueType
Dequeue
Dequeue<'a> is the consuming side of a queue. Every Queue is a Dequeue, and
Hub.subscribe returns a bare Dequeue, so one consumer works for
both, and a subscriber cannot offer into its own buffer.
take waits for the next value; takeBetween waits for a
minimum and takes a batch; poll, takeUpTo, and
takeAll never wait. Values leave in FIFO order even when takers are interrupted: waiting
takers are served one at a time, so a value given back by an interrupted taker returns before anything later is taken.
shutdown ends a queue without discarding its backlog: later takes drain what remains, and
on a hub subscription it unsubscribes. stats reads the backlog and the accepted, dropped, and
evicted counters for monitoring.
Consume a queue as a stream with FlowStream.fromDequeue.
The Queue guide covers batching, shutdown, and monitoring.
Specification
Summary
| Name | Signature | Synopsis |
|---|---|---|
| Operations | ||
| take | Dequeue.take queue | Removes the oldest value, suspending until one is available. |
| poll | Dequeue.poll queue | Removes the oldest value if one is available, without suspending. |
| takeUpTo | Dequeue.takeUpTo max queue | Removes up to max available values in FIFO order, without suspending. |
| takeBetween | Dequeue.takeBetween min max queue | Removes between min and max values in FIFO order, suspending until at least min are available. |
| takeAll | Dequeue.takeAll queue | Removes every available value in FIFO order, without suspending. |
| size | Dequeue.size queue | Returns the number of buffered values. |
| stats | Dequeue.stats queue | Reads the queue's backlog and counters in one consistent snapshot. |
| capacity | Dequeue.capacity queue | Returns the queue's capacity, or None for an unbounded queue. |
| shutdown | Dequeue.shutdown queue | Shuts the queue down. |
| isShutdown | Dequeue.isShutdown queue | Returns whether the queue has been shut down. |
| awaitShutdown | Dequeue.awaitShutdown queue | Suspends until the queue is shut down. |
Operations
Parameters
| Name | Type | Description |
|---|---|---|
| queue | Dequeue<'a> |
Returns
Verification Examples
flow {
let! (jobs: Queue<string>) = Queue.bounded 8
do! jobs |> Queue.offer "job" |> Flow.ignore
return! Dequeue.take jobs
}min and max values in FIFO order, suspending until at
least min are available.
Parameters
| Name | Type | Description |
|---|---|---|
| min | int | |
| max | int | |
| queue | Dequeue<'a> |
Returns
Verification Examples
// Wait for at least one sample, then write up to 500 in one batch.
samples |> Dequeue.takeBetween 1 500Parameters
| Name | Type | Description |
|---|---|---|
| queue | Dequeue<'a> |
Returns
Verification Examples
flow {
let! stats = Dequeue.stats historian
if stats.Capacity |> Option.exists (fun limit -> stats.Size * 10 >= limit * 8) then
do! raiseAlarm "historian buffer 80% full"
}